package ntd.util;

import cn.hutool.core.lang.Console;
import cn.hutool.core.util.RandomUtil;
import cn.hutool.core.util.StrUtil;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * 模拟一个客户端接收消息
 *
 * @author Unclue_liu
 */
public class UtilMqttClient {

    private String HOST;
    private String clientId;
    private MqttClient client;
    private MqttConnectOptions options;
    private String userName;    //非必须
    private String passWord;  //非必须

    private static Map<String, UtilMqttClient> utilMqttClientMap = new HashMap();

    /**
     * 以流程为单位进行mqtt服务存储
     */
    public static Map<String, List<UtilMqttClient>> mqttServers = new HashMap<>();

    private UtilMqttClient(String HOST, String clientId, String userName, String passWord) throws MqttException {
        this.HOST = HOST;
        this.clientId = clientId;
        this.userName = userName;
        this.passWord = passWord;
        init();
    }

    public static UtilMqttClient getClient(String HOST, String clientId, String userName, String passWord) throws MqttException {
        if(StrUtil.isBlank(clientId)){
            clientId = RandomUtil.randomNumbers(12);
        }
        UtilMqttClient utilMqttClient = utilMqttClientMap.get(clientId);
        if(utilMqttClient == null){
            utilMqttClient = new UtilMqttClient(HOST, clientId, userName, passWord);
            utilMqttClientMap.put(clientId, utilMqttClient);
        }
        return utilMqttClient;
    }

    private void init() throws MqttException {
        // host为主机名，clientid即连接MQTT的客户端ID，一般以唯一标识符表示，MemoryPersistence设置clientid的保存形式，默认为以内存保存
        client = new MqttClient(HOST, clientId, new MemoryPersistence());
        // MQTT的连接设置
        options = new MqttConnectOptions();
        // 设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录，设置为true表示每次连接到服务器都以新的身份连接
        options.setCleanSession(false);
        if(StrUtil.isAllNotBlank(userName, passWord)){
            options.setUserName(userName); // 设置连接的用户名
            options.setPassword(passWord.toCharArray()); // 设置连接的密码
        }
        // 设置超时时间 单位为秒
        options.setConnectionTimeout(10);
        // 设置会话心跳时间 单位为秒 服务器会每隔1.5*20秒的时间向客户端发送个消息判断客户端是否在线，但这个方法并没有重连的机制
        options.setKeepAliveInterval(20);
        //设置断开后重新连接
        options.setAutomaticReconnect(true);
        client.connect(options);
    }

    public MqttTopic subscribe(String topic, MqttCallback mqttCallback, int qos) throws MqttException {
        // 设置回调
        client.setCallback(mqttCallback);
        MqttTopic mqttTopic = client.getTopic(topic);
        //setWill方法，如果项目中需要知道客户端是否掉线可以调用该方法。设置最终端口的通知消息
        options.setWill(mqttTopic, "close".getBytes(), 1, true); //遗嘱
        //订阅消息
        int[] Qos = {qos};//0：最多一次 、1：最少一次 、2：只有一次
        String[] topic1 = { topic };
        client.subscribe(topic1, Qos);
        return mqttTopic;
    }

    public void sendMessage(String topic, String msg, int qos) {
        sendMessage(topic, msg, qos, false);
    }

    public void sendMessage(String topic, String msg, int qos, boolean retain) {
        Console.log("发送消息 - 开始" , msg);
        final MqttMessage mqttMessage = new MqttMessage();
        mqttMessage.setQos(qos);  //保证消息能到达一次
        mqttMessage.setRetained(retain);
        mqttMessage.setPayload(("{\"mag\":\"" + msg + "\"}").getBytes());
        try {
            final MqttTopic mqttTopic = client.getTopic(topic);
            MqttDeliveryToken token = mqttTopic.publish(mqttMessage);
            token.waitForCompletion();
            Console.log("发送消息 - 完成");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 关闭mqtt
     */
    public void close (){
        try {
            client.close(true);
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

    /**
     * 保存mqtt服务状态
     * @param flowId
     * @param utilMqttClient
     */
    public static void push(String flowId, UtilMqttClient utilMqttClient){
        List<UtilMqttClient> utilMqttClients = mqttServers.get(flowId);
        if(utilMqttClients == null){
            utilMqttClients = new ArrayList<>();
            mqttServers.put(flowId, utilMqttClients);
        }
        utilMqttClients.add(utilMqttClient);
    }

    /**
     * 清理某个流程的mqttServer服务
     * @param flowId
     */
    public static void clearMqttServer(String flowId){
        List<UtilMqttClient> utilMqttClients = mqttServers.get(flowId);
        if(utilMqttClients == null || utilMqttClients.size() < 1){
            return;
        }
        utilMqttClients.forEach(utilMqttClient -> {
            utilMqttClient.close();
        });
        mqttServers.remove(flowId);
    }

    public static void main(String[] args) throws MqttException {
        UtilMqttClient client2 = new UtilMqttClient("tcp://127.0.0.1:1883", "server", null, null);
        client2.sendMessage("topic-1", "测试", 1, false);

        UtilMqttClient client = new UtilMqttClient("tcp://127.0.0.1:1883", "client", null, null);
        client.subscribe("topic-1", new MqttCallback() {
            @Override
            public void connectionLost(Throwable throwable) {
                // 连接丢失后，一般在这里面进行重连
                System.out.println("连接断开，可以做重连");
            }
            @Override
            public void messageArrived(String topic, MqttMessage mqttMessage) {
                // subscribe后得到的消息会执行到这里面
                System.out.println("接收消息主题 : " + topic);
                System.out.println("接收消息Qos : " + mqttMessage.getQos());
                System.out.println("接收消息内容 : " + new String(mqttMessage.getPayload()));
            }
            @Override
            public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
                System.out.println("deliveryComplete---------" + iMqttDeliveryToken.isComplete());
            }
        }, 1);

    }
}
